package promote.brozen.rsocket.channel;

import io.netty.buffer.Unpooled;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.core.Resume;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;

import java.util.Scanner;

/**
 * @author Brozen
 * @date 2020/6/28 9:41 AM
 * @email brozen@qq.com
 */
public class ChannelClient {

    public static void main(String[] args) {
        TcpClientTransport clientTransport = TcpClientTransport.create("localhost", 8009);

        RSocket rsocket = RSocketConnector.create()
                // Resume重连，当客户端与服务端断开连接时，可以尝试重连，要求客户端与服务端都开启此项配置才生效
                // 在客户端，需要设置一个提供token提供者，否则在resume的时候服务端会返回错误: unknown resume token
                // 这个Resume只能处理网络问题导致的连接断开，
                // 如果是服务端挂掉了，服务端的SessionManager中不再保留所有客户端数据，重连也找不到Session
                // 如果是客户端挂掉了，服务端SessionManager还存有会话，客户端可以设置token来resume，但是
                .resume(new Resume().token(() -> {
                    byte[] token = "brozen".getBytes();
                    return Unpooled.wrappedBuffer(token);
                }))
//                .reconnect(Retry.backoff(20, Duration.ofMillis(500)))
                .connect(clientTransport)
                .block();

        RoleController roleController = new RoleController();
        // warning 如果不想自己在Publisher的onSubscribe中重写Subscription
        //         必须用Flux.just然后concat拼接publisher，才能成功发送数据
        //         onSubscribe中不重写Subscription就直接用Subscriber发射数据不能成功
        rsocket.requestChannel(
                roleController)
                /*Flux.just(DefaultPayload.create("1wefca", "124ew"))
                        .concatWith(roleController))*/
                // 返回的flux如果直接subscribe，会出现重复订阅报错？貌似也只是debug级别的报错，没有影响，也可以正常使用
                // 但从官方demo来看，是用doOnXXX的方法去接收Payload
                .doOnNext(roleController::onNext)
                .doOnError(roleController::onError)
                .doOnComplete(roleController::onComplete)
                .subscribe();

        rsocket.onClose().subscribe(payload -> {
            System.out.println("client closed");
        });

        rsocket.onClose().block();
    }


    private static class RoleController extends BaseSubscriber<Payload> implements Publisher<Payload> {

        private String username;

        private Scanner scanner;

        private Subscriber<? super Payload> subscriber;

        private Thread thread;

        public RoleController() {
            this(new Scanner(System.in));
        }

        public RoleController(Scanner scanner) {
            this.scanner = scanner;

            System.out.print("请输入用户名：");
            this.username = scanner.nextLine();
        }

        @Override
        public void subscribe(Subscriber<? super Payload> subscriber) {
            this.subscriber = subscriber;
            this.subscriber.onSubscribe(new Subscription() {
                @Override
                public void request(long n) {
                    if (n == 1) {
                        // n == 1 表示是第一个请求，需要发送鉴权Payload
                        subscriber.onNext(DefaultPayload.create(username, username));
                    } else {
                        // 鉴权Payload验证通过后，会请求一个 n == Integer.MAX_VALUE，这时候就可以正常发送数据了
                        start();
                    }
                }

                @Override
                public void cancel() {
                    thread.interrupt();
                }
            });
        }

        public void start() {
            this.thread = new Thread(() -> {
                String msg = scanner.nextLine();
                while (!"exit".equals(msg)) {
                    Payload payload = DefaultPayload.create(msg, username);
                    System.out.println("[Client Send]" + msg);

                    if (this.thread.isInterrupted()) {
                        break;
                    }

                    this.subscriber.onNext(payload);
                    msg = scanner.nextLine();
                }
            });
            this.thread.start();
        }

        @Override
        protected void hookOnNext(Payload payload) {
            System.out.println("[client receive] " + payload.getDataUtf8());
        }

        @Override
        protected void hookOnError(Throwable t) {
            t.printStackTrace();
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("[client] complete");
        }
    }
}
